package org.shj.spark.streaming

import org.apache.spark.streaming.StreamingContext
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.streaming.Seconds

object FileStreamDemo {
  def main(args: Array[String]): Unit = {
    if (args.length < 1) {
      System.err.println("Usage: HdfsWordCount <directory>")
      System.exit(1)
    }
    
    val conf = new SparkConf().setAppName("SocketStreamDemo")
    conf.setMaster("local")
    val sc = new SparkContext(conf)
    //sc.setLogLevel("WARN")
    
    val ssc = new StreamingContext(sc, Seconds(10)) 
    
    val directory = args(0)
    
    /**
     * 下面方法中的KeyClass, ValueClass, InputFormatClass 都不知道是啥
     * 此方法会读取目录下的所有文件，当目录有新文件生成时，该文件也会被处理。
     * 注意： 1. 子目录不会被处理
     *      2. 目录下的文件需要有相同的data format
     *      3. 文件被处理过后，如果又在文件中追加了新内容，追加的内容不会被处理
     */
    //val what = ssc.fileStream[KeyClass, ValueClass, InputFormatClass](directory)
    val lines = ssc.textFileStream(directory)
    
    val pairs = lines.flatMap(_.split("\\s+")).map(word => (word, 1))
    
    val wordCnt = pairs.reduceByKey(_+_)
    
    wordCnt.print()
    
    ssc.start()
    ssc.awaitTermination()
    
  }
}